#include "helper/rocksdb/Table.h"

#include <iostream>

#include "helper/rocksdb/SliceUtil.h"
#include "helper/rocksdb/Options.h"

#include "rocksdb/db.h"


namespace helper {
namespace rocksdb {

Table::Table(std::shared_ptr<::rocksdb::DB> db, const std::string & name, ::rocksdb::ColumnFamilyHandle * handle, const Options & options)
:_db(db)
, _name(name)
, _handle(handle)
, _options(options)
{
}
Table::~Table(){
    _handle = nullptr;
    _db.reset();
}

::rocksdb::Status Table::get(const Slice & key, std::string & value) {
	return _db->Get(_options.read(), _handle, key, &value);
}
::rocksdb::Status Table::put(const Slice & key, const Slice & value){
	return _db->Put(_options.write(), _handle, key, value);
}
::rocksdb::Status Table::del(const Slice & key){
	return _db->Delete(_options.write(), _handle, key);
}

::rocksdb::Status Table::get(const int64_t key, std::string & value) {
	return _db->Get(_options.read(), _handle, SliceUtil::fromNumber(key), &value);
}
::rocksdb::Status Table::put(const int64_t key, const Slice & value){
	return _db->Put(_options.write(), _handle, SliceUtil::fromNumber(key), value);
}
::rocksdb::Status Table::del(const int64_t key){
	return _db->Delete(_options.write(), _handle, SliceUtil::fromNumber(key));
}

::rocksdb::Status Table::get(const uint64_t key, std::string & value) {
	return _db->Get(_options.read(), _handle, SliceUtil::fromNumber(key), &value);
}
::rocksdb::Status Table::put(const uint64_t key, const Slice & value){
	return _db->Put(_options.write(), _handle, SliceUtil::fromNumber(key), value);
}
::rocksdb::Status Table::del(const uint64_t key){
	return _db->Delete(_options.write(), _handle, SliceUtil::fromNumber(key));
}

bool Table::foreach(const Visitor & v, const Slice & start, const Slice & end) const{
	std::unique_ptr<::rocksdb::Iterator> iter(_db->NewIterator(_options.read(), _handle));
	if(iter == nullptr){
		std::cerr << "ERROR: new iterator null" << std::endl;
		return false;	
	}

    if(start == Slice()){
        iter->SeekToFirst();
    }
    else{
        iter->Seek(start);
    }

	for(; iter->Valid() && _options.comparator().Compare(iter->key(), end) != 0; iter->Next()){
        if(!v(iter->key(), iter->value())) return false;
	}
	return true;
}

}//end namespace rocksdb
}//end namespace helper
